package cn._51doit.flink.day07;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * 获取EventTime类型窗口的迟到数据
 * <p>
 * 默认的情况，迟到的数据是被丢弃的
 */
public class GetWindowLateData {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //1000,spark,3
        //3000,hadoop,5
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //提取数据中的eventTime
        SingleOutputStreamOperator<String> dataStreamWithWaterMark = lines.assignTimestampsAndWatermarks(
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String ele, long l) {
                        String[] fields = ele.split(",");
                        return Long.parseLong(fields[0]);
                    }
                }));

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = dataStreamWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String line) throws Exception {
                String[] fields = line.split(",");
                String word = fields[1];
                int count = Integer.parseInt(fields[2]);
                return Tuple2.of(word, count);
            }
        });

        //先KeyBy
        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(tp -> tp.f0);

        OutputTag<Tuple2<String, Integer>> lateTag = new OutputTag<Tuple2<String, Integer>>("latData") {
        };

        //按照EventTime滚动的窗口，窗口长度为10秒
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyed
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .sideOutputLateData(lateTag) //给迟到的数据打标签
                ;

        SingleOutputStreamOperator<Tuple2<String, Integer>> res = window.sum(1);

        //获取窗口迟到的数据
        DataStream<Tuple2<String, Integer>> lateDataStream = res.getSideOutput(lateTag);

        res.print();

        lateDataStream.print("late ");

        env.execute();

    }


}
